package org.onion.pattern.event.outbox.executor;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.onion.pattern.event.outbox.OutboxDeliveryExecutor;
import org.onion.pattern.event.outbox.config.EventOutboxConfig;
import org.onion.pattern.event.outbox.data.OutboxEventInfo;

import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;


@Slf4j
@Setter
@Getter
public abstract class GroupingOutboxDeliveryExecutor<E> implements OutboxDeliveryExecutor {

    private volatile boolean stop     = false;
    /**
     * 轮询间隔(毫秒)
     */
    @Setter
    private volatile int     interval = 100;

    @Setter
    private volatile int     batchSize = 100; //一批查询个数
    @Setter
    private volatile int     retry     = 3;
    private volatile boolean started   = false;

    private final Map<String, Exec> execPool = new ConcurrentHashMap<>(100);

    public GroupingOutboxDeliveryExecutor(EventOutboxConfig config) {
        setRetry(ObjectUtils.defaultIfNull(config.getRetryOnPublishFail(), 3));
        setBatchSize(ObjectUtils.defaultIfNull(config.getBatchSize(), 100));
        setInterval(ObjectUtils.defaultIfNull(config.getInterval(), 100));

        //是否开启投递
        Boolean deliveryEnable = ObjectUtils.defaultIfNull(config.getDeliverEnabled(), true);
        if (deliveryEnable) {
            Set<String> groupSet = ObjectUtils.defaultIfNull(config.getGroups(), Collections.emptySet());
            for (String p : groupSet) {
                p = StringUtils.trimToNull(p);
                if (p == null) {
                    continue;
                }
                Exec exec = new Exec(p);
                execPool.put(p, exec);
            }
            //添加默认分区
            if (!execPool.containsKey(OutboxEventInfo.NONE_GROUP)) {
                execPool.put(OutboxEventInfo.NONE_GROUP, new Exec(OutboxEventInfo.NONE_GROUP));
            }
        }

    }

    @Override
    public synchronized void start() {
        if (this.started) {
            return;
        }

        log.info("【OUTBOX-事件分发】 启动");
        this.started = true;
        for (Exec exec : execPool.values()) {
            /**
             * 守护线程的生命周期依赖于创建它的用户线程，当所有非守护线程结束时，JVM会自动退出，不会等待守护线程完成。
             * 因此，守护线程适合执行一些后台任务，如垃圾回收、应用指标统计等。但是，它们不应该用于执行对数据完整性有要求的操作，因为守护线程可能会在没有完成任务的情况下被JVM强制结束。
             * @date 2025-08-02 0:23
             */
//            exec.setDaemon(true); // 设置为守护线程
            exec.start();
        }
    }

    @Override
    public void flush() {
        for (Exec value : execPool.values()) {
            value.activate();
        }
    }

    public void flush(Set<String> groupSet) {
        if (CollectionUtils.isEmpty(groupSet)) {
            return;
        }

        for (String group : groupSet) {
            Exec exec = execPool.get(group);
            if (exec == null) {
                continue;
            }
            exec.activate();
        }
    }

    @Override
    public synchronized void shutdown() {
        if (stop) {
            return;
        }
        log.info("【OUTBOX-事件分发】关闭");
        stop = true;
        for (Exec value : execPool.values()) {
            value.interrupt();
        }
        execPool.clear();
    }

    protected abstract ExecStatus runWrapper(ExecStatus status, Function<ExecStatus, ExecStatus> fun);

    /**
     * 查询就绪的事件
     */
    protected abstract List<E> findReadyEvent(String group, int limit) throws Exception;

    /**
     * 发送事件
     */
    protected abstract void publish(List<E> msg) throws Exception;

    /**
     * 确认已发送的事件
     */
    protected abstract boolean ack(List<E> event);


    @Setter
    @Getter
    private class Exec extends Thread {

        private final Semaphore semaphore = new Semaphore(1);
        private final String    group;
        private       List<E>   ackEvents = new ArrayList<>(100);
        private final Random    random    = new Random();

        public Exec(String group) {
            super("EventDeliveryThread:" + group);
            this.group = StringUtils.defaultIfBlank(group, OutboxEventInfo.NONE_GROUP);
        }

        /**
         * 使用活动起来
         */
        void activate() {
            this.semaphore.drainPermits();
            this.semaphore.release(1);
        }

        /**
         * 随机数,如果 interval 值是 200 。
         */
        private int random() {

            return Math.max(random.nextInt(interval), 100);

//            // 生成高斯分布随机数，范围在2-6秒之间
//            double mean = 4.0;  // 均值（2和6的中间值）
//            double stdDev = 0.67;  // 标准差，调整这个值可以控制分布的宽度
//
//            // 生成随机数
//            double gaussianValue;
//            do {
//                gaussianValue = mean + random.nextGaussian() * stdDev;
//            } while (gaussianValue < 2.0 || gaussianValue > 6.0);  // 确保在2-6范围内

        }

        @Override
        public void run() {
            ExecStatus execStatus = ExecStatus.CONTINUABLE;
            log.info("【OUTBOX-事件分发】 启动线组 group: {}  time:{}", group, LocalDateTime.now());

            while (!stop) {

                //在间隔
                try {
                    if (ExecStatus.INTERVAL == execStatus) {
                        this.semaphore.tryAcquire(1, random(), TimeUnit.MILLISECONDS);
                    }
                } catch (InterruptedException e) {
                    log.info("【OUTBOX-事件分发】 Thread interrupted while waiting for lock");
                    Thread.currentThread().interrupt(); // 恢复中断状态
                    break;
                }

                execStatus = runWrapper(execStatus, (stats) -> {
                    if (stop && ackEvents.isEmpty()) {
                        return ExecStatus.STOP;
                    }

                    //查询就绪的事件
                    List<E> eventInfos = Collections.emptyList();
                    try {
                        if (ackEvents.isEmpty()) { //还有没ack 的事件,就不需要查询了
                            eventInfos = findReadyEvent(group, batchSize);
                            log.trace("【OUTBOX-事件分发】 group {}  查询记录数 {}", group, eventInfos.size());
                            if (eventInfos.isEmpty()) {
                                return ExecStatus.INTERVAL;
                            }
                        }
                    } catch (Throwable e) {
                        log.error("【OUTBOX-事件分发】 查询事件失败", e);
                        return ExecStatus.INTERVAL;
                    }

                    if (stop && ackEvents.isEmpty()) {
                        return ExecStatus.STOP;
                    }

                    //发送事件
                    boolean fail = true;
                    for (int i = 0; i < retry; i++) {
                        if (!ackEvents.isEmpty()) { //还有没ack 的事件,就不需要发送了
                            fail = false;
                            break;
                        }

                        try {
                            publish(eventInfos);
                            ackEvents.addAll(eventInfos);
                            fail = false;
                            break;
                        } catch (Throwable e) {
                            log.error("【OUTBOX-事件分发】 发送事件失败", e);
                        }
                    }

                    if (fail) {
                        return ExecStatus.INTERVAL;
                    }

                    //消息确认
                    boolean success = ack(ackEvents);

                    if (!success) { //ack 失败
                        return ExecStatus.CONTINUABLE;
                    }

                    ackEvents.clear();

                    if (eventInfos.size() >= batchSize) {
                        return ExecStatus.CONTINUABLE; //本批次不是最后一批
                    }

                    return ExecStatus.INTERVAL;
                });

                if (execStatus == ExecStatus.STOP) {
                    break;
                }
            }

            log.warn("【OUTBOX-事件分发】退出  group: {}  {}", group, LocalDateTime.now());
        }

    }

    protected enum ExecStatus {
        /**
         * 停止
         */
        STOP,

        /**
         * 继续
         */
        CONTINUABLE,

        /**
         * 间隔
         */
        INTERVAL
    }
}
